package com.lazyben.client;

import com.lazyben.util.SpringUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Scope("prototype")
@Slf4j
@Getter
@Component
public class MyClient implements Runnable {
    @Value("${netty.server.port}")
    private int port; // 端口

    @Value("${netty.server.host}")
    private String host; // push server IP

    private ChannelFuture channelFuture;

    private final String clientId; // 移动设备识别码 International Mobile Equipment Identity

    public MyClient(String clientId) {
        this.clientId = clientId;
    }

    @Override
    public void run() {
        try {
            startClient();
            log.info("客户端启动成功，clientId:{}", this.getClientId());
        } catch (InterruptedException e) {
            log.error("客户端启动失败:{}", e.getMessage(), e);
        }
    }

    public void close() {
        if (null != this.channelFuture) {
            this.channelFuture.channel().close();
        }
    }

    private void startClient() throws InterruptedException {
        EventLoopGroup eventExecutors = new NioEventLoopGroup();
        ClientHandler clientHandler = SpringUtils.getBean(ClientHandler.class, this);
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventExecutors)
                 .channel(NioSocketChannel.class)
                 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                 .option(ChannelOption.SO_RCVBUF, 1024 * 32)
                 .option(ChannelOption.SO_SNDBUF, 1024 * 32)
                 .handler(new ChannelInitializer<SocketChannel>() {
                     @Override
                     protected void initChannel(SocketChannel socketChannel) throws Exception {

//                         socketChannel.pipeline().addLast(clientHandler);
                         socketChannel.pipeline()
                                      .addLast(new DelimiterBasedFrameDecoder(1024 * 1024, Unpooled.copiedBuffer("\r\n".getBytes())))
//                                      .addLast(new StringEncoder(CharsetUtil.UTF_8)) // String编码。
//                                      .addLast(new StringDecoder(CharsetUtil.UTF_8)) // String解码。
                                      .addLast("ping", new IdleStateHandler(0, 0, 20, TimeUnit.SECONDS))
                                      .addLast(clientHandler); // 心跳设置

                     }
                 });
        connect(bootstrap); // 连接服务端
    }

    private void connect(Bootstrap b) throws InterruptedException {
        long c1 = System.currentTimeMillis();
        final int maxRetries = 2; //重连2次
        final AtomicInteger count = new AtomicInteger();
        final AtomicBoolean flag = new AtomicBoolean(false);
        try {
            this.channelFuture = b.connect(host, port).addListener(
                    new ChannelFutureListener() {
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if (!future.isSuccess()) {
                                if (count.incrementAndGet() > maxRetries) {
                                    log.warn("clientId={}重连超过{}次", clientId, maxRetries);
                                } else {
                                    log.info("clientId={}重连第{}次", clientId, count);
                                    b.connect(host, port).addListener(this);
                                }

                            } else {
                                log.info("clientId={}连接成功,连接IP:{}连接端口:{}", clientId, host, port);
                                flag.set(true);
                            }
                        }
                    }).sync(); //同步连接
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        if (flag.get()) {
            log.info("设备clientId={}，channelId={}连接耗时={}ms", clientId, channelFuture.channel().id(), System.currentTimeMillis() - c1);
            channelFuture.channel().closeFuture().sync(); //连接成功后将持续阻塞该线程
        }
    }
}
